package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.estimateSoh.ConfigData;
import cn.gwm.flink.streaming.beans.estimateSoh.EstimateSohRecord;
import cn.gwm.flink.streaming.constant.CellBatterTypeEnum;
import cn.gwm.flink.streaming.constant.DateTimeBucketFormat;
import cn.gwm.flink.streaming.function.reduce.EstimateSohReduceFunction;
import cn.gwm.flink.streaming.strategy.estimatesoh.OcvStrategy;
import cn.gwm.flink.streaming.strategy.estimatesoh.OcvStrategyFactory;
import cn.gwm.utils.StreamUtil;
import cn.hutool.core.date.CalendarUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;

import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.json.JSONUtil;

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;


/**
 * 自定义有状态的mapFunction
 *
 * @author GW00283474
 */
public class EstimateSohProcessFunction extends KeyedProcessFunction<String, EstimateSohRecord, EstimateSohRecord> {
    private static final Logger logger= LoggerFactory.getLogger(EstimateSohProcessFunction.class);

    /**
     * 休眠时间
     */
    private ValueState<Long> sleepTimeState;
    /**
     * 上电时间
     */
    private ValueState<Long> powerOnTimeState;
    /**
     * 休眠静止状态是否检查过一次
     */
    private ValueState<Boolean> checkDormantStaticState;
    /**
     * 1800秒中记录的集合
     */
    private ListState<EstimateSohRecord> estimateSohRecordListState;
    /**
     * 定时器集合
     */
    private MapState<Long,Long> timersState;
    /**
     * 计算电池累积充电容量
     */
    private ReducingState<BigDecimal> cumulativeChrgAhRmState;

    /**
     * 计算电池累积放电容量
     */
    private ReducingState<BigDecimal> cumulativeDchrgAhRmState;

    /**
     * 静止休眠上电时间与上一次休眠时间的时间间隔(秒)
     */
    static final Integer STATIC_DORMANT_DV = 3600;
    /**
     * 在线静止休眠电流持续时长属性
     */
    static final String  BATT_CURR_DURATION= "battCurrDuration";
    /**
     * 在线静置判断条件二 电流绝对值大于等于
     */
    static final String BATT_CURR_MIN = "battCurrMin";

    /**
     * 在线静置判断条件二 电流绝对值小于等于
     */
    static final String BATT_CURR_MAX = "battCurrMax";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // MapState 状态的TTL 是针对状态的处理时间定义 （ProcessingTime）
        StateTtlConfig stateTtlConfig = StateTtlConfig
                // 状态有效时间    1天过期
                .newBuilder(Time.days(1))
                // 设置状态的更新类型
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 已过期还未被清理掉的状态数据不返回给用户
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                // 过期对象的清理策略 全量清理
                .cleanupFullSnapshot()
                .build();
        ValueStateDescriptor descriptor = new ValueStateDescriptor("sleepTimeDesc",Long.class);
        descriptor.enableTimeToLive(stateTtlConfig);
        ValueStateDescriptor powerOnTimeDesc = new ValueStateDescriptor("powerOnTimeDesc",Long.class);
        powerOnTimeDesc.enableTimeToLive(stateTtlConfig);
        ValueStateDescriptor checkDormantStaticStateDesc = new ValueStateDescriptor("checkDormantStaticStateDesc",Boolean.class);
        checkDormantStaticStateDesc.enableTimeToLive(stateTtlConfig);
        ReducingStateDescriptor cumulativeChrgAhRmStateDesc = new ReducingStateDescriptor("cumulativeChrgAhRmStateDesc",new EstimateSohReduceFunction(),BigDecimal.class);
        ReducingStateDescriptor cumulativeDchrgAhRmStateDesc = new ReducingStateDescriptor("cumulativeDchrgAhRmStateDesc",new EstimateSohReduceFunction(),BigDecimal.class);
        ListStateDescriptor  estimateSohListStateDesc = new ListStateDescriptor("estimateSohListStateDesc",EstimateSohRecord.class);
        estimateSohListStateDesc.enableTimeToLive(stateTtlConfig);

        MapStateDescriptor  timersStateDesc = new MapStateDescriptor("timersStateDesc",Long.class,Long.class);
        timersStateDesc.enableTimeToLive(stateTtlConfig);
        sleepTimeState = getRuntimeContext().getState(descriptor);
        powerOnTimeState = getRuntimeContext().getState(powerOnTimeDesc);
        checkDormantStaticState = getRuntimeContext().getState(checkDormantStaticStateDesc);
        estimateSohRecordListState= getRuntimeContext().getListState(estimateSohListStateDesc);
        timersState=getRuntimeContext().getMapState(timersStateDesc);
        cumulativeChrgAhRmState=getRuntimeContext().getReducingState(cumulativeChrgAhRmStateDesc);
        cumulativeDchrgAhRmState=getRuntimeContext().getReducingState(cumulativeDchrgAhRmStateDesc);

    }

    @Override
    public void processElement(EstimateSohRecord record, KeyedProcessFunction<String, EstimateSohRecord, EstimateSohRecord>.Context ctx, Collector<EstimateSohRecord> out) throws Exception {//检查休眠静止状态
        try {
            record.setProcessDate(DateUtil.format(CalendarUtil.calendar().getTime(), DateTimeBucketFormat.YYYYMMdd));
            checkDormantStaticStatus(record);
            //检查OCV非平台有效性
            checkFlatValid(record);
            //注册触发器
            //在线静止休眠电流持续时长1800秒
            long battCurrDuration=0L;
            ConfigData configDataBattCurrDuration=record.getConfigDataMap().get(BATT_CURR_DURATION);
            if(Objects.nonNull(configDataBattCurrDuration)){
                battCurrDuration=Long.parseLong(configDataBattCurrDuration.getConfigValue());
            }
            long timer=record.getItem_time()+(battCurrDuration*1000L);
            //System.out.println("vin:"+record.getVin()+",信号时间戳："+DateUtil.format(CalendarUtil.calendar(ctx.timestamp()).getTime(), DatePattern.NORM_DATETIME_PATTERN)+",水位线："+ctx.timerService().currentWatermark()+",tid:"+DateUtil.format(CalendarUtil.calendar(record.getTid()).getTime(), DatePattern.NORM_DATETIME_PATTERN));
            //System.out.println("vin:"+record.getVin()+",计算在线静止状态起始时间："+DateUtil.format(new Date(timer), DatePattern.NORM_DATETIME_PATTERN));
            //System.out.println("注册定时器："+DateUtil.format(CalendarUtil.calendar(timer).getTime(), DatePattern.NORM_DATETIME_PATTERN));
            ctx.timerService().registerEventTimeTimer(timer);
            timersState.put(record.getItem_time(),timer);
            //收集记录信息
            estimateSohRecordListState.add(record);
            //计算电池累积充、放电容量
            countBmsTotlAhRm(record);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, EstimateSohRecord, EstimateSohRecord>.OnTimerContext ctx, Collector<EstimateSohRecord> out) throws Exception {
        try {
            //System.out.println("vin:"+ctx.getCurrentKey()+",computerTime："+DateUtil.format(CalendarUtil.calendar(timestamp).getTime(), DatePattern.NORM_DATETIME_PATTERN));
            Iterator<EstimateSohRecord> recordIterator = estimateSohRecordListState.get().iterator();
            List<EstimateSohRecord> recordList = StreamUtil.convertIterator(recordIterator)
                    .sorted(new Comparator<EstimateSohRecord>() {
                        @Override
                        public int compare(EstimateSohRecord o1, EstimateSohRecord o2) {
                            return (int) (o1.getItem_time()- o2.getItem_time());
                        }
                    }).collect(Collectors.toList());
            EstimateSohRecord first=recordList.stream().findFirst().orElse(null);
            if(Objects.nonNull(first)){
                Stream<EstimateSohRecord> handleRecordStream = recordList.stream().filter(s -> BooleanUtil.and(s.getItem_time() >= first.getItem_time(), s.getItem_time() < timestamp));
                List<EstimateSohRecord> estimateSohRecords = handleRecordStream.collect(Collectors.toList());
                //条件一 整车状态为Ready （vuc=2）
                boolean vcuMatch = estimateSohRecords.stream().anyMatch(record -> 2 != record.getVCU_VCUSts());
                if(!vcuMatch){
                    //条件二 1800秒内电流区间【0A,20A】
                    boolean match = estimateSohRecords.stream().allMatch(record -> {
                        double bmsBattCurrRmAbs = Math.abs(record.getBMS_BattCurr_RM().doubleValue());
                        double battCurrMin=0.0d;
                        double battCurrMax=20.0d;
                        ConfigData configDataMin = record.getConfigDataMap().get(BATT_CURR_MIN);
                        ConfigData configDataMax = record.getConfigDataMap().get(BATT_CURR_MAX);
                        if(Objects.nonNull(configDataMin)){
                            battCurrMin= Double.parseDouble(configDataMin.getConfigValue());
                        }
                        if(Objects.nonNull(configDataMax)){
                            battCurrMax= Double.parseDouble(configDataMax.getConfigValue());
                        }
                        return BooleanUtil.and(bmsBattCurrRmAbs >= battCurrMin, bmsBattCurrRmAbs <= battCurrMax);
                    });
                    if(match){
                        first.setOnlineStillFlag(true);
                    }
                }
/*            StreamUtil.convertIterator(timersState.values().iterator()).forEach(s-> {
                try {
                    System.out.println("定时器列表："+DateUtil.format(CalendarUtil.calendar(s).getTime(), DatePattern.NORM_DATETIME_PATTERN));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });*/
                ctx.timerService().deleteEventTimeTimer(timersState.get(first.getItem_time()));
                long firstTimer=timersState.get(first.getItem_time());
                timersState.remove(first.getItem_time());
                //System.out.println("移除定时器："+DateUtil.format(CalendarUtil.calendar(firstTimer).getTime(), DatePattern.NORM_DATETIME_PATTERN));
                //更新1800秒累积的信号信息
                Stream<EstimateSohRecord> stream = StreamUtil.convertIterator(estimateSohRecordListState.get().iterator());
                List<EstimateSohRecord> list=stream.filter(s->NumberUtil.isGreater(new BigDecimal(String.valueOf(s.getItem_time())),new BigDecimal(String.valueOf(first.getItem_time())))).collect(Collectors.toList());
                estimateSohRecordListState.update(list);
                //检查触发条件
                checkTriggerFlag(first);
                //检查使能条件
                checkEnableFlag(first);
                out.collect(first);
            }
            //移除已触发的定时器
            if(Objects.nonNull(timestamp)){
                ctx.timerService().deleteEventTimeTimer(timestamp);
            }
        }catch (Exception e){
            logger.error("---------------定时器计算过程中出现异常message:{}",e.getMessage());
            logger.error("---------------定时器计算过程中出现异常Cause:{}",e.getCause());
            logger.error("---------------定时器计算过程中出现异常StackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }


    }

    /**
     * 检查ocv平台有效性
     */
    void checkFlatValid(EstimateSohRecord record){
        try {
            OcvStrategy ocvStrategy = OcvStrategyFactory.getOcvStrategy(CellBatterTypeEnum.BATTER_TYPE_LFP.getKeyByValue(record.getCell_battery_type()));
            ocvStrategy.execute(record);
            if(BooleanUtil.or(record.getPlatformValidFlagMin(),record.getPlatformValidFlagMax())){
                record.setPlatformValidFlag(true);
            }
        }catch (Exception e){
            logger.error("检查ocv平台有效性异常message:{}",e.getMessage());
            logger.error("检查ocv平台有效性异常cause:{}",e.getCause());
            logger.error("检查ocv平台有效性异常StackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }

    }
    /**
     * 检查休眠静止状态
     *
     * @param record
     * @throws Exception
     */
    void checkDormantStaticStatus(EstimateSohRecord record){
        try {
            //vcuVcuSts 0:休眠   1：上电
            //暂时未加VCU_VCUSts信号防抖 1分钟
            if (0 == record.getVCU_VCUSts()) {
                checkDormantStaticState.clear();
                sleepTimeState.update(NumberUtil.max(record.getItem_time(), null == sleepTimeState.value()? 0 : sleepTimeState.value()));
            } else {
                if (1 == record.getVCU_VCUSts() && Objects.nonNull(sleepTimeState.value())) {
                    if (Objects.isNull(checkDormantStaticState.value()) && Objects.nonNull(powerOnTimeState.value()) && Objects.nonNull(sleepTimeState.value())) {
                        long dv = DateUtil.between(DateUtil.date(powerOnTimeState.value()), DateUtil.date(sleepTimeState.value()), DateUnit.SECOND);
                        if (dv >= STATIC_DORMANT_DV) {
                            record.setSleepStillFlag(true);
                            checkDormantStaticState.update(true);
                        }
                    } else {
                        record.setSleepStillFlag(true);
                    }
                    //上电信号时间
                    powerOnTimeState.update(NumberUtil.max(record.getItem_time(), null == powerOnTimeState.value()? 0 : powerOnTimeState.value()));
                }
            }
        }catch (Exception e){
            logger.error("-------检查休眠静止状态:{}",JSONUtil.toJsonStr(record));
            logger.error("检查休眠静止状态异常message:{}",e.getMessage());
            logger.error("检查休眠静止状态异常case:{}",e.getCause());
            logger.error("检查休眠静止状态异常stackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }

    }

    /**
     * 检查触发条件
     * @param record
     */
    void checkTriggerFlag(EstimateSohRecord record){
        try {
            if(BooleanUtil.or(record.getSleepStillFlag(),record.getOnlineStillFlag())){
                record.setTriggerFlag(true);
            }
        }catch (Exception e){
            logger.error("---------检查触发条件:{}",JSONUtil.toJsonStr(record));
            logger.error("检查触发条件异常message:{}",e.getMessage());
            logger.error("检查触发条件异常cause:{}",e.getCause());
            logger.error("检查触发条件异常StackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }

    }

    /**
     * 检查使能条件
     * @param record
     */
    void checkEnableFlag(EstimateSohRecord record){
        try {
            if(BooleanUtil.and(record.getTriggerFlag(),record.getPlatformValidFlagMax())){
                record.setEnableFlag(true);
            }
        }catch (Exception e){
            logger.error("---------检查使能条件:{}",JSONUtil.toJsonStr(record));
            logger.error("检查使能条件异常message:{}",e.getMessage());
            logger.error("检查使能条件异常cause:{}",e.getCause());
            logger.error("检查使能条件异常stackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }

        //System.out.println("-----------record.getTriggerFlag():"+record.getTriggerFlag()+"------"+"record.getPlatformValidFlagMax():"+record.getPlatformValidFlagMax()+"---"+"record.getEnableFlag():"+record.getEnableFlag()+"--->");

    }

    /**
     * 计算电池累积充、放电容量
     * @param record
     */
    void countBmsTotlAhRm(EstimateSohRecord record) {
        try {
            if(11==record.getVCU_VCUSts()){
                BigDecimal currentBmsTotlChargAhRm=record.getBMS_TotlChrgAh_RM();
                BigDecimal currentBmsTotlDhargAhRm=record.getBMS_TotlDchrgAh_RM();
                if(Objects.nonNull(currentBmsTotlChargAhRm)){
                    cumulativeChrgAhRmState.add(currentBmsTotlChargAhRm);
                }
                record.setCumulativeChrgAhRm(null==cumulativeChrgAhRmState.get()?new BigDecimal("0"):cumulativeChrgAhRmState.get());
                if(Objects.nonNull(currentBmsTotlDhargAhRm)){
                    cumulativeDchrgAhRmState.add(currentBmsTotlDhargAhRm);
                }
                record.setCumulativeDchrgAhRm(null==cumulativeDchrgAhRmState.get()?new BigDecimal("0"):cumulativeDchrgAhRmState.get());
            }
            if(Objects.nonNull(record.getCumulativeChrgAhRm())&&Objects.nonNull(record.getCumulativeDchrgAhRm())){
                record.setTotlAh(record.getCumulativeChrgAhRm().subtract(record.getCumulativeDchrgAhRm()));
            }
        }catch (Exception e){
            logger.error("--------计算电池累积充、放电容量:{}",JSONUtil.toJsonStr(record));
            logger.error("计算电池累积充、放电容量异常message:{}",e.getMessage());
            logger.error("计算电池累积充、放电容量异常cause:{}",e.getCause());
            logger.error("计算电池累积充、放电容量异常stackTrace:{}",e.getStackTrace());
            e.printStackTrace();
        }

    }
}
